package com.ruoyi.flink.source;/**
 * @program: ruoyi
 * @description: CustomSourceFunciton
 * @author: zengchen
 * @create: 2020-12-06 10:44
 **/

import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @program: ruoyi
 * @description: CustomSourceFunciton
 * @author: zengchen
 * @create: 2020-12-06 10:44
 **/
public class CustomSourceFunction1 implements SourceFunction<Long> {

    private long count = 1L;
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext ctx) throws Exception {
        while (isRunning && count <= 100) {
            synchronized (ctx.getCheckpointLock()) {
                ctx.collect(count);
                count++;
                Thread.sleep(2000);
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
